package com.example.yckjbigdataflink;

import com.example.yckjbigdataflink.sink.HBaseUpdateSink;
import com.example.yckjbigdataflink.source.HBaseScanSource;
import com.example.yckjbigdataflink.utils.HBaseConfigurationUtil;
import com.example.yckjbigdataflink.utils.KerberosUtil;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;

public class MainFlinkApp {
    public static void main(String[] args) throws Exception {
        // 获取配置
        ParameterTool params = ParameterTool.fromArgs(args);
        String krb5ConfPath = params.getRequired("krb5Conf");
        String keytabPath = params.getRequired("keytab");
        String principal = params.getRequired("principal");
        String hbaseTable = params.getRequired("hbaseTable");
        String hbaseZkQuorum = params.getRequired("hbaseZkQuorum");
        
        // 从配置获取参数
        /*
        "use command as: \n" + "java -cp /opt/FI-Client/Flink/flink/lib/*:/opt/FlinkConfigtableJavaExample.jar"
                + " com.huawei.bigdata.flink.examples.RedisDataImport --configPath <config filePath>"
         */
        final String configureFilePath = ParameterTool.fromArgs(args).get("configPath", "config/import.properties");
        final String csvFilePath =
                ParameterTool.fromPropertiesFile(configureFilePath).get("CsvPath", "config/configtable.csv");
        final boolean isHasHeaders =
                ParameterTool.fromPropertiesFile(configureFilePath).getBoolean("CsvHeaderExist", true);
        final String csvScheme = ParameterTool.fromPropertiesFile(configureFilePath).get("ColumnNames");
        final boolean isSecurity =
                ParameterTool.fromPropertiesFile(configureFilePath).getBoolean("Redis_Security", true);
        final String redisIPPort = ParameterTool.fromPropertiesFile(configureFilePath).get("Redis_IP_Port");


        // 1. Kerberos登录
        KerberosUtil.login(krb5ConfPath, keytabPath, principal);


        // 2. 设置HBase配置
        org.apache.hadoop.conf.Configuration hbaseConf = HBaseConfigurationUtil.getHBaseConfiguration(hbaseZkQuorum);
        hbaseConf.set("hbase.security.authentication", "kerberos");
        hbaseConf.set("hbase.master.kerberos.principal", "hbase/_HOST@YOUR.REALM");
        hbaseConf.set("hbase.regionserver.kerberos.principal", "hbase/_HOST@YOUR.REALM");


        // 3. 初始化Flink环境
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();


        // 每10秒触发一次检查点： https://mp.weixin.qq.com/s/b3_gEga_GWho14thPnCemQ
        env.enableCheckpointing(10000);
        // 获取检查点配置
        CheckpointConfig checkpointConfig = env.getCheckpointConfig();
        // 设置模式为EXACTLY_ONCE （EXACTLY_ONCE/AT_LEAST_ONCE）
        checkpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        // 设置检查点超时时间
        checkpointConfig.setCheckpointTimeout(300000);
        // 设置最小间隔时间
        checkpointConfig.setMinPauseBetweenCheckpoints(1000);
        // 设置最大并发检查点数
        checkpointConfig.setMaxConcurrentCheckpoints(1);
        // 设置外部化检查点的清理行为
        checkpointConfig.setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        // 设置检查点存储
        env.setStateBackend(new FsStateBackend("hdfs:///flink/checkpoints"));


        env.setParallelism(20); // 并行数 = HBase region 数或更小 (全局）。自上而下：执行环境的并行度被设置为3，这意味着如果没有在算子层次显式设置并行度，所有算子的并行度都将为3。
        // 单独为算子设置并行度 sum(1)算子的并行度被设置为5，这意味着该算子会启动5个并发任务来处理数据
        // 执行环境层次: Flink程序运行在执行环境的上下文中，执行环境为所有执行的算子、数据源、数据接收器（sink）定义了一个默认的并行度。
        // 可以显式配置算子层次的并行度去覆盖执行环境的并行度。可以通过调用 setParallelism() 方法指定执行环境的默认并行度。
        /*DataStream<String> text =[...];
        DataStream<Tuple2<String,Integer>> wordCounts = text
                .flatMap(newLineSplitter())
                .keyBy(value -> value.f0)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .sum(1).setParallelism(5);*/
        // 客户端层次: 将作业提交到Flink时可在客户端设定其并行度。对于CLI客户端，可以通过 -p 参数指定并行度。
        // ./bin/flink run -p 10 WordCount-java.jar
        // 1.2.4 系统层次
        // 可以通过设置 ./conf/flink-conf.yaml 文件中的 parallelism.default 参数，在系统层次来指定所有执行环境的默认并行度。
        // parallelism.default:2

        // Slot的配置方式
        // 2.3.1 全局配置
        // 在 flink-conf.yaml 文件中设置 taskmanager.numberOfTaskSlots 参数，为整个集群设定默认的Task Slot数量。
        // taskmanager.numberOfTaskSlots:4
        // 2.3.2 命令行参数
        // 在启动Flink集群时，使用 -D taskmanager.numberOfTaskSlots=4 参数覆盖默认值。
        // ./bin/start-cluster.sh -D taskmanager.numberOfTaskSlots=4
        // 2.3.3 动态配置
        // 对于某些特定的应用场景，可能需要更灵活地控制每个TaskManager的Task Slot数量。这时可以利用Flink提供的REST API或者YARN/Kubernetes等平台提供的机制来进行动态调整。

        // 算子的子任务与Slot的分配
        //      在Flink中，每个算子会根据其并行度被拆分成多个子任务（subtask），这些子任务需要被分配到不同的Slot中执行。
        //      例如，一个算子的并行度为3，
        //      那么它会有3个子任务，这3个子任务需要被分配到3个不同的Slot中。默认情况下，Flink允许子任务共享Slot，
        //      只要这些子任务属于同一作业。这样可以提高资源利用率，例如将资源密集型和非密集型的任务同时放到一个Slot中，
        //      它们可以自行分配对资源占用的比例。

        // 算子链对Slot使用的影响
        //      算子链（Operator Chain）是Flink中的一种优化技术，它将多个算子连接在一起形成一个链式结构，
        //      以减少数据序列化和网络传输开销。当算子形成算子链后，它们会被合并成一个任务，这个任务只需要一个Slot来执行。
        //      例如，Source算子和Map算子形成了算子链，它们的并行度都为2，那么合并后的任务也有2个子任务，只需要2个Slot来执行。

        // 算子间的数据传输模式
        // 3.2.1 一对一（One-to-one，forwarding）
        //      这种模式下，数据流维护着分区以及元素的顺序。比如source和map算子，source算子读取数据之后，
        //      可以直接发送给map算子做处理，它们
        //      之间不需要重新分区，也不需要调整数据的顺序。这就意味着map算子的子任务，看到的元素个数和顺序跟source
        //      算子的子任务产生的完全一样，保证着“一对一”的关系。map、filter、flatMap 等算子都是这种one-to-one的对应关系。
        //      这种关系类似于Spark中的窄依赖。
        // 3.2.2 重分区（Redistributing）
        //      在这种模式下，数据流的分区会发生改变。比如 map 和后面的 keyBy/window 算子之间，以及 keyBy/window
        //       算子和 Sink 算子之间，都是这样的关系。每一个算子的子任务，会根据数据传输的策略，把数据发送到不同的
        //      下游目标任务。这些传输方式都会引起重分区的过程，这一过程类似于Spark中的shuffle。
        // 3.3 算子链的创建条件
        //      3.3.1 上下游的并行度一致: 上下游算子的并行度必须相同，才能形成算子链。例如，如果Source算子的并行度为2，Map算子的并行度为4，那么它们之间就不能形成算子链。
        //      3.3.2 下游节点的入度为1: 下游节点的入度为1，即下游节点没有来自其他节点的输入。例如，如果一个算子有两个输入流，那么它就不能与上游算子形成算子链。
        //      3.3.3 上下游节点都在同一个slot group中: 上下游节点都必须在同一个slot group中，才能形成算子链。关于slot group的概念，将在后面的共享组部分详细介绍。
        //      3.3.4 下游节点的chain策略为ALWAYS: 下游节点的chain策略为ALWAYS，表示可以与上下游链接，map、flatmap、filter 等默认是ALWAYS。
        //      3.3.5 上游节点的chain策略为ALWAYS或HEAD: 上游节点的chain策略为ALWAYS或HEAD，HEAD表示只能与下游链接，不能与上游链接，Source默认是HEAD。
        //      3.3.6 两个节点间数据分区方式是forward: 两个节点间的数据分区方式必须是forward，即数据不需要重新分区。
        //      3.3.7 用户没有禁用chain: 用户没有通过编程API禁用算子链。例如，没有调用 disableChaining() 方法。
        // 3.4 算子链的控制方法
        //      3.4.1 全局禁用算子链
        //      StreamExecutionEnvironment env =StreamExecutionEnvironment.getExecutionEnvironment();
        //      env.disableOperatorChaining();

        //      3.4.2 从当前算子开始新链: 可以通过在DataStream的operator后面调用 startNewChain() 来指示从该operator开始一个新的chain（与前面截断，不会被chain到前面）。示例代码如下：
        //      DataStream<String> stream = env.fromElements("a","b","c");
        //      stream.map(value -> value.toUpperCase())
        //      .startNewChain()
        //      .filter(value -> value.startsWith("A"));

        //      3.4.3 禁用算子链:可以通过调用 disableChaining() 来指示该operator不参与chaining（不会与前后的operator chain一起）。示例代码如下：
        //      DataStream<String> stream = env.fromElements("a","b","c");
        //      stream.map(value -> value.toUpperCase())
        //      .disableChaining()
        //      .filter(value -> value.startsWith("A"));

        //  考虑算子的资源需求:
        //      对于资源密集型的算子，如aggregate、reduce、sum、window等，即使并行度相同，也可以考虑不与其他算子形成算子链，
        //      而是单独分配Slot，以确保其有足够的资源执行。例如，对于一个复杂的窗口操作，可以使用 startNewChain() 
        //      或 disableChaining() 方法将其与其他算子分开，使其独享一个Slot的资源。而对于非资源密集型的算子，
        //      如source、map、sink等，可以尽量与其他算子形成算子链，共享Slot资源。

        // 共享组的代码示例: 在Apache Flink中，slotSharingGroup() 是一个用于控制算子（operator）之间资源共享的机制。
        // 它允许多个算子共享相同的slot（即资源容器）。Slot是Flink中的资源单位，slot共享可以提高资源利用率，但在某些情况下，
        // 我们希望更精细地控制不同算子的资源分配，slotSharingGroup 就提供了这种能力。
        //  解决背压问题: 对于某些复杂的算子，可能会导致算子链中的其他算子受到背压影响。通过将其分配到不同的slot sharing group，可以减少此类问题。例如，对于一个计算密集型的算子，可以将其分配到一个独立的slot sharing group，避免对其他算子产生背压。
        //// 定义两个数据流
        //DataStream<String> stream1 = env.fromElements("a","b","c");
        //DataStream<String> stream2 = env.fromElements("1","2","3");
        //// 给第一个算子链设置 slotSharingGroup
        //stream1.map(value -> value.toUpperCase())
        //.slotSharingGroup("group1")
        //.filter(value -> value.startsWith("A"))
        //.slotSharingGroup("group1");
        //// 给第二个算子链设置不同的 slotSharingGroup
        //stream2.map(value -> value +"X")
        //.slotSharingGroup("group2")
        //.filter(value -> value.endsWith("X"))
        //.slotSharingGroup("group2");
        //// 汇聚两个流并继续处理
        //stream1.union(stream2)
        //.map(value ->"Processed: "+ value)
        //.slotSharingGroup("group3");
        //env.execute();
        // 在上述代码中，stream1 的算子被分配到了 "group1"，stream2 的算子被分配到了 "group2"
        // ，两者之间的算子不会共享相同的slot，从而实现了资源隔离。最后，通过 union() 操作将两个流合并并
        // 设置为 "group3"，合并后的流将使用一个新的共享组。

        // env.fromSequence(1, 10);

        // 若任务仅为 ETL 扫描任务，可禁用或开启异步快照：
        //env.enableCheckpointing(60000);
        //env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

        // 4. 构造HBase数据源（官方HBase connector示例）
        DataStream<Result> hbaseStream = env.addSource(new HBaseScanSource());
        // 4. 构造HBase数据源（官方HBase connector示例）
        /*HBaseTableSource hBaseSource = new HBaseTableSource(hbaseConf, hbaseTable);
        DataStreamSource<Result> hbaseStream = env.createInput(hBaseSource);*/

        // 5. 解析HBase Result 转换成你的业务数据
        /*hbaseStream
                .map((MapFunction<Result, String>) result -> {
                    // 假设读取rowKey 和某个cf:qualifier列
                    String rowKey = Bytes.toString(result.getRow());
                    byte[] val = result.getValue(Bytes.toBytes("cf"), Bytes.toBytes("qualifier"));
                    String value = val != null ? Bytes.toString(val) : "";
                    return rowKey + ":" + value;
                })
                // 6. Sink写入HDFS 或 Hive （示例写文本到HDFS）
                .writeAsText("hdfs://namenode:8020/user/flink/output/", org.apache.flink.core.fs.FileSystem.WriteMode.OVERWRITE);
*/
        // 6.1.1 keyBy 之前发生数据倾斜. 可以使用下面任意一种.例如 Kafka 的 topic 中某些 partition 的数据量比较大，
        // 某些 partition 的数据量比较少。
        // 对于不存在 keyBy 的 Flink 任务也会出现数据倾斜的情况。可以使用 shuffle、rebalance 或 rescale 算子将数据均匀分配
        /*DataStream<Result> resultDataStream = hbaseStream
                .rebalance()
                .shuffle()
                .rescale();*/

        // 6.1.2 keyby 后的聚合操作存在数据倾斜
        // 使用 LocalKeyBy 的思想，在 keyBy 上游算子数据发送之前，首先在上游算子的本地对数据进行聚合后再发送到下游，
        // 使下游接收到的数据量大大减少，从而使得 keyBy 之后的聚合操作不再是任务的瓶颈。具体实现方式为：
        // keyby 之前使用 flatMap 实现 LocalKeyBy。当数据达到 x 条或者 x 时间后，批次数据保存在状态后端，聚合一次后输出一条。
        /*DataStream<Event> localAggregatedStream= inputStream
                .flatMap(newLocalKeyByFlatMap())
                .keyBy(Event::getKey)
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                .aggregate(newMyAggregateFunction());*/

        // 6.1.3 keyby 后的窗口聚合操作存在数据倾斜
        // 因为使用了窗口，变成了有界数据的处理，窗口默认时触发关窗时才会输出一条结果发往下游，所以可以使用两阶段聚合的方式。
        // 第一阶段聚合：key 拼接随机数前缀或后缀，进行 keyby、开窗、聚合。聚合完不再是 WindowedStream，
        // 要获取 WindowEnd 作为窗口标记作为第二阶段分组依据，避免不同窗口的结果聚合到一起。
        // 第二阶段聚合：去掉随机数前缀或后缀，按照原来的 key 及 windowStart/windowEnd 作为 keyby、聚合。
        /*DataStream<Event>inputStream=...;
        DataStream<Event>firstStageAggregatedStream=inputStream
                .map(newAddRandomPrefixMap())
                .keyBy(Event::getKeyWithPrefix)
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                .aggregate(newFirstStageAggregateFunction());

        DataStream<Event>finalAggregatedStream=firstStageAggregatedStream
                .map(newRemoveRandomPrefixMap())
                .keyBy(Event::getKey)
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                .aggregate(newSecondStageAggregateFunction());*/

        // 6.2 算子性能优化
        // 6.2.2 异步 I/O 操作
        // 对于与外部系统交互频繁的算子，如 Sink 节点写入数据库、Lookup Join 等，可以使用异步 I/O 操作，减少等待时间，
        // 提高处理速度。Flink 提供了 AsyncFunction 接口，方便实现异步 I/O 操作。
        /*DataStream<String>inputStream=...;
        AsyncDataStream.unorderedWait(inputStream,newAsyncDatabaseLookupFunction(),1000,TimeUnit.MILLISECONDS,100);*/

        // 转换数据 示例：解析 Result -> POJO
        DataStream<MyRow> dataStream = hbaseStream.map(result -> {
            String rowkey = Bytes.toString(result.getRow());
            String name = Bytes.toString(result.getValue(Bytes.toBytes("cf"), Bytes.toBytes("name")));
            int age = Bytes.toInt(result.getValue(Bytes.toBytes("cf"), Bytes.toBytes("age")));
            return new MyRow(rowkey, name, age);
        });
        //.print(); // 不打印
        dataStream.addSink(new HBaseUpdateSink());
        // 6.5.2 批量操作
        // 在与外部系统交互时，尽量使用批量操作，减少单次操作的开销。例如，在写入数据库时，使用批量写入的方式，提高写入效率。
        // inputStream.addSink(newBatchDatabaseSink());

        // Flink 1.13 以后的版本，Flink Web UI 的监控中，通过颜色加数值，更清晰明了地表明每个算子的繁忙程度和反压程度。
        // 正常情况下为蓝色 ->
        // 紫色 -> 黑色 -> 淡红 -> 红，繁忙和反压程度逐渐加深。同时，为每个算子提供了 SubTask 级别的 BackPressure 监控，
        // 更便于观察该节点是否处于反压状态。默认情况下，0.1 表示 OK，0.1 - 0.5 表示 LOW，超过 0.5 表示 HIGH。如果出现反压，
        // 7.2 自动反压保护
        // Flink 提供了自动反压保护机制，可以通过设置setAutoWatermarkInterval来调整。当检测到反压时，
        // Flink 会自动减慢数据源的发送速度，直到下游处理速度跟上。
        /*
        DataStream<String>stream=...;// 获取输入数据流
        stream
                .setAutoWatermarkInterval(1000L)// 设置自动反压保护的间隔为 1 秒
                .addSink(...);// 设置数据输出*/

        env.execute("Flink Read Billion Rows from HBase");
    }

    public static class MyRow {
        public String getOp() {
            return op;
        }

        public void setOp(String op) {
            this.op = op;
        }

        private String op;
        public String id;

        public String getId() {
            return id;
        }

        public void setId(String id) {
            this.id = id;
        }

        public String getName() {
            return name;
        }

        public void setName(String name) {
            this.name = name;
        }

        public String name;

        public int getAge() {
            return age;
        }

        public void setAge(int age) {
            this.age = age;
        }

        public int age;
        public MyRow() {}
        public MyRow(String id, String name, int age) {
            this.id = id; this.name = name; this.age = age;
        }
        @Override public String toString() {
            return String.format("MyRow{id=%s, name=%s, age=%d}", id, name, age);
        }
    }
}
